package org.orbis.memorymanager;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.msgpack.object.ArrayType;
import org.orbis.communicate.FlagsResult;
import org.orbis.communicate.MainService;
import org.orbis.communicate.deleter.DeleterPackage;
import org.orbis.communicate.indexer.term.TermPackage;
import org.orbis.constants.Constants;
import org.orbis.flag.SearchFlags;
import org.orbis.logging.LogParser;
import org.orbis.memoryblock.appendix.AppendixPackage;
import org.orbis.memoryblock.content.ContentPackage;
import org.orbis.memoryblock.dictionary.DictionaryPackage;
import org.orbis.memorycontroller.DiskController;
import org.orbis.memorycontroller.MemoryController;
import org.orbis.memorycontroller.struct.MemoryInfoMaterial;
import org.orbis.memorymanager.filter.DeleteFilter;
import org.orbis.search.SearchPackage;
import org.orbis.search.SearchResult;
import org.orbis.structure.ContentInfo;
import org.orbis.structure.LookupPointer;

public class MemoryManager{
	private static final int DIGLENGTH = Constants.SKIPPTR_INTERVAL;
	private static final int MEM_SKIPPTR = Constants.MEMORY_TYPE_SKIPPTR;
	// private static final int MEM_SKIPCONT = Constants.MEMORY_TYPE_SKIPPTRCONT;

	// Memory / Disk controller
	private MemoryController memory;
	private DiskController disk;
	private MainService mains;

	public boolean stopFlag = false;

	private int termcounter = 0;

	private ContentInfo contentInfo;
	private int intFlagNum;
	private long longFlagNum;

	private Lock termLock = new ReentrantLock();
	private Lock invertLock = new ReentrantLock();

	private DeleteFilter delFilter;

	public MemoryManager(int blockSize, int blockLines, String dirPath, int backupInterval, int replication,
		int total, int cluster, int uniqueKeyIndex, MainService mains){
		this.mains = mains;
		this.disk = new DiskController(dirPath, blockSize);
		this.contentInfo = mains.getConf().contentInfo();
		this.intFlagNum = contentInfo.integerFlagNum();
		this.longFlagNum = contentInfo.longFlagNum();
		try{
			this.delFilter = new DeleteFilter(mains.getConf());
		}catch(IOException e){
			e.printStackTrace();
		}
		this.memory =
			new MemoryController(blockSize, blockLines, disk, contentInfo, backupInterval, replication,
				total, cluster, uniqueKeyIndex, delFilter, this);

		LogParser.fine("Memory manager started.");
	}

	public void storeTerm(TermPackage pkg){
		try{
			// Remove package's ID from deleted list
			delFilter.removeID(pkg.ID());
			for(int i = 0; i < contentInfo.integerFlagNum(); i++){
				delFilter.removeIntFlag(i, pkg.intFlag(i));
			}
			for(int i = 0; i < contentInfo.longFlagNum(); i++){
				delFilter.removeLongFlag(i, pkg.longFlag(i));
			}

			TermInformation newDic = null;
			try{
				termLock.lock();
				// Store term to dictionary, skip if the term exists
				newDic = storeDictionary(pkg.term(), pkg.valueType(), pkg.cluster(), pkg.termFreq());
			}finally{
				termLock.unlock();
			}

			// store invert index and IDs
			try{
				invertLock.lock();
				storeInvIndex(newDic.getDictionary().getCurrent(), pkg);
			}finally{
				invertLock.unlock();
			}
		}catch(Exception e){
			e.printStackTrace();
		}
		if(termcounter++ % 100000 == 0) System.gc();
	}

	// Method to store content
	public void storeContent(String content, long ID, int index, LookupPointer[] appendix, int cluster){
		ContentInformation contTerm;

		contTerm = lookupContentPointer(ID, index, cluster);
		if(contTerm == null) contTerm = ContentInformation.getDefault().ID(ID);

		// System.out.println("[DEBUG] Previous: " + contTerm.getPrevious() + " Current: "
		// + contTerm.getCurrent() + " Next: " + contTerm.getNext());

		ContentPackage pkg = ContentPackage.setPackage(appendix, content).ID(ID).cluster(cluster);
		LookupPointer storePtr = memory.storeContent(pkg, index);
		contTerm.current(storePtr);

		// System.out.println("[DEBUG2] Previous: " + contTerm.getPrevious() + " Current: "
		// + contTerm.getCurrent() + " Next: " + contTerm.getNext());

		if(!contTerm.getPrevious().equals(LookupPointer.EMPTY))
			memory.storeContentNextPointer(storePtr, contTerm.getPrevious());

		makeSkipPtrForContent(contTerm, ID, index);
	}

	// pick up pointer to invert indices
	public TermInformation searchTerm(String term, boolean[] valueTypes, int cluster){
		TermInformation newTerm = lookupTerm(term, cluster);
		// lookup term
		if(newTerm.getDictionary() == null) return newTerm;

		// Filter
		for(int i = 0; i < contentInfo.contentsNum(); i++)
			if(valueTypes[i] == false) newTerm.deleteInvIndex(i);

		return newTerm;
	}

	public FlagsResult[] getInvertIndex(LookupPointer ptr, SearchPackage searchPkg){
		if(ptr == null) return null;
		FlagsResult[] pkg = memory.lookupInvertIndex(ptr);
		FlagsResult[] filtered = new FlagsResult[pkg.length];
		int count = 0;
		boolean isBreak;
		SearchFlags ranges = searchPkg.flags();
		// delete by ID or flag
		for(int i = 0; i < pkg.length; i++){
			isBreak = false;
			long nextOne = pkg[i].ID();
			boolean[] boolFlags = pkg[i].binaryFlags();
			int[] intFlags = pkg[i].intFlags();
			long[] longFlags = pkg[i].longFlags();
			float[] floatFlags = pkg[i].floatFlags();

			// Delete
			if(delFilter.filterID(nextOne)) continue;
			for(int j = 0; j < intFlagNum; j++){
				if(delFilter.filterIntFlag(j, intFlags[j])){
					isBreak = true;
					break;
				}
			}
			for(int j = 0; j < longFlagNum; j++){
				if(delFilter.filterLongFlag(j, longFlags[j])){
					isBreak = true;
					break;
				}
			}

			// Range filter
			int rangeLength = ranges.getBinaryFlags().length;
			for(int j = 0; j < rangeLength; j++){
				if(!ranges.inquiryBinary(j, boolFlags[j])){
					isBreak = true;
					break;
				}
			}
			rangeLength = ranges.getIntRange().length;
			for(int j = 0; j < rangeLength; j++){
				if(!ranges.inquiryInt(j, intFlags[j])){
					isBreak = true;
					break;
				}
			}
			rangeLength = ranges.getLongRange().length;
			for(int j = 0; j < rangeLength; j++){
				if(!ranges.inquiryLong(j, longFlags[j])){
					isBreak = true;
					break;
				}
			}
			rangeLength = ranges.getFloatRange().length;
			for(int j = 0; j < rangeLength; j++){
				if(!ranges.inquiryFloat(j, floatFlags[j])){
					isBreak = true;
					break;
				}
			}
			if(isBreak) continue;
			filtered[count++] = pkg[i];
		}

		return Arrays.copyOf(filtered, count);
	}

	public SearchResult searchContent(long ID, int cluster){
		int i;
		LookupPointer priorPtr = null;
		String[] contents = new String[contentInfo.contentsNum()];
		String[] appendices = new String[contentInfo.appendixNum()];

		for(i = 0; i < contentInfo.contentsNum(); i++){
			ContentInformation ptrInfo = lookupContentPointer(ID, i, cluster);
			if(ptrInfo.getCurrent() != null){
				if(i == 0) priorPtr = ptrInfo.getCurrent();
				String content = memory.lookupContents(ptrInfo.getCurrent());
				contents[i] = content;
				if(delFilter.contentIndex() == i && delFilter.filterContent(contents[i])) return null;
			}else{
				System.out.printf("Cannot found ID %d, prev: %d-%s next: %d-%s%n", ID, cluster,
					ptrInfo.getPrevious(), cluster, ptrInfo.getNext());
				return null;
			}
		}

		// Delete filter
		if(priorPtr != null){
			LookupPointer[] appendixPointers = memory.lookupContentToAppendixPointer(priorPtr);
			for(i = 0; i < appendixPointers.length; i++){
				appendices[i] = memory.lookupAppendix(appendixPointers[i]).getString();
				if(delFilter.appendixIndex() == i && delFilter.filterAppendix(appendices[i])) return null;
			}
		}
		SearchResult result = SearchResult.makeResult(ID, 0.0, contents, appendices);

		return result;
	}

	public int getTerms(){
		return termcounter;
	}

	public ArrayList<MemoryInfoMaterial> getStatusList(){
		return memory.getStatusList();
	}

	public void getMemoryList(){
		memory.showMemoryElapsed();
	}

	public void initializeMemory(){
		try{
			memory.readMemory();
		}catch(Exception ignored){}
	}

	public void saveMemory(){
		memory.saveToMemory();
	}

	public int getSkipPointerLen(int cluster){
		return memory.getSkipPointerLength(cluster);
	}

	private TermInformation lookupTerm(String query, int cluster){
		LogParser.finest("Find around \"" + query + "\"");
		// Look at skip pointer first
		int skipPointerLength = memory.getSkipPointerLength(cluster);
		LookupPointer dicPtr = null;
		DictionaryPackage storedTerm = null;
		TermInformation returnTerm = null;
		int i, skipLevel = 0, skipVal = 1;

		if(skipPointerLength > 1){
			skipLevel = (int)(Math.log(skipPointerLength - 1) / Math.log(DIGLENGTH));
			skipVal = (int)Math.pow(DIGLENGTH, skipLevel);
		}
		// Find from pointers in skip-pointer, if not, state to traversal search
		for(i = 0; i < skipPointerLength; i += skipVal){
			dicPtr = memory.lookupSkipPointer(i, cluster);
			// No pointer
			if(dicPtr == null) break;

			storedTerm = memory.lookupDictionary(dicPtr);
			int comp = storedTerm.getName().compareTo(query);
			LogParser.finest("Comparing \"" + storedTerm.getName() + "\" Result: " + comp);
			if(comp > 0){
				// skip pointer advances -> revert pointer
				skipLevel--;
				if(skipLevel == -1 || i == 0){
					i--;
					break;
				}
				i -= skipVal;
				skipVal = (int)Math.pow(DIGLENGTH, skipLevel);
			}else if(comp == 0) return new TermInformation(storedTerm).skipptr(i);

			// traverse
			if(i + skipVal >= skipPointerLength){
				do{
					skipLevel--;
					if(skipLevel == -1){
						skipVal = 1;
						break;
					}
					skipVal = (int)Math.pow(DIGLENGTH, skipLevel);
				}while(i + skipVal >= skipPointerLength);
			}
		}
		if(i == -1){
			// prior to the first pointer
			if(storedTerm != null) dicPtr = storedTerm.getCurrent();
			storedTerm = null;
		}else{
			// get skip pointer again
			if(i == skipPointerLength) i--;
			if(i >= 0){
				dicPtr = memory.lookupSkipPointer(i, cluster);
				storedTerm = memory.lookupDictionary(dicPtr);
			}
		}

		// find by dictionary order
		if(storedTerm != null){
			LookupPointer prevPtr = storedTerm.getCurrent();
			DictionaryPackage nextTerm = null;
			if(storedTerm.getNext() != null) nextTerm = memory.lookupDictionary(storedTerm.getNext());
			int count = 0;
			while(nextTerm != null){
				LogParser.finest("Comparing " + nextTerm.getName());
				int comp = nextTerm.getName().compareTo(query);
				if(comp > 0){
					// Not exists
					returnTerm =
						new TermInformation(null).middle(prevPtr, nextTerm.getCurrent()).skipptr(i)
							.appendSkip(count >= DIGLENGTH);
					break;
				}else if(comp == 0){
					// Found term in the dictionary
					returnTerm = new TermInformation(nextTerm).middle(prevPtr, nextTerm.getNext()).skipptr(i);
					break;
				}else{
					prevPtr = nextTerm.getCurrent();
					if(nextTerm.getNext() == null) break;
					nextTerm = memory.lookupDictionary(nextTerm.getNext());
				}
				count++;
			}

			if(returnTerm == null){
				// The final term in dictionary
				returnTerm =
					new TermInformation(null).behindLast(prevPtr).skipptr(i).appendSkip(count >= DIGLENGTH);
			}
		}else{
			// prior to first term(i = -1) or the first term(i = 0)
			returnTerm = new TermInformation(null).behindFirst(dicPtr).skipptr(i);
		}

		return returnTerm;
	}

	private void makeSkipPtrForTerm(TermInformation term){
		int index = term.getSkipIndex();

		if(index == -1){
			// Overwrite first pointer
			memory.storeSkipPointer(0, -1, term.getDictionary().getCurrent(), true, MEM_SKIPPTR, term
				.getDictionary().getCluster());
			refreshSkipPtrDict(term.getDictionary().getCluster());
		}else{
			if(term.isAppendSkipPointer()){
				// Append pointer
				memory.storeSkipPointer(index + 1, -1, term.getDictionary().getCurrent(), false, MEM_SKIPPTR,
					term.getDictionary().getCluster());
			}
		}
	}

	private void makeSkipPtrForContent(ContentInformation term, long storeID, int type){
		int index = term.getSkipIndex();

		if(index == -1){
			// overwrite first pointer
			refreshSkipPtrContent(type, term.getID(), term.getCurrent(), term.getCurrent().cluster());
		}else{
			if(term.isAppendSkipPointer()){
				// append skip pointer
				memory.storeContentSkipPointer(type, storeID, term.getCurrent(), term.getCurrent().cluster());
			}
		}
	}

	private void refreshSkipPtrDict(int cluster){
		LookupPointer threshold = memory.lookupSkipPointer(1, cluster);
		LookupPointer nextPtr = memory.lookupSkipPointer(0, cluster);
		LookupPointer prevPtr = null;
		int count = 0;
		if(threshold == null){
			while(nextPtr != null){
				prevPtr = nextPtr;
				nextPtr = memory.lookupDictionaryNext(nextPtr);
				count++;
			}
		}else{
			while(nextPtr != null && !threshold.equals(nextPtr)){
				prevPtr = nextPtr;
				nextPtr = memory.lookupDictionaryNext(nextPtr);
				count++;
			}
		}

		if(count >= DIGLENGTH) memory.storeSkipPointer(1, 0, prevPtr, false, MEM_SKIPPTR, cluster);
	}

	// refresh skip pointer not to jam in the first pointer
	private void refreshSkipPtrContent(int type, long ID, LookupPointer pointer, int cluster){
		memory.modifyContentSkipPointer(type, ID, pointer, cluster, 0);
	}

	// Store an term
	private TermInformation storeDictionary(String query, int valueType, int cluster, int frequency)
		throws UnsupportedEncodingException{
		TermInformation newTerm = lookupTerm(query, cluster);

		if(!newTerm.isFound()){
			// Add term if a query doesn't appear
			DictionaryPackage addPkg =
				DictionaryPackage.setPackage(query, contentInfo.contentsNum(), cluster)
					.next(newTerm.getNext()).freq(frequency);
			LookupPointer newTermPtr = memory.storeDictionary(addPkg, cluster);
			newTerm.dictionary(addPkg.current(newTermPtr));

			// Reconnect pointer to the term
			if(newTerm.getPrevious() != null)
				memory.storeDictionaryPointer(newTerm.getPrevious(), newTermPtr, cluster);

			// Add and modify dictionary
			makeSkipPtrForTerm(newTerm);
		}else{
			memory.incrementDocFreq(newTerm.getDictionary().getCurrent(), cluster);
		}

		return newTerm;
	}

	private void storeInvIndex(LookupPointer dicPtr, TermPackage pkg){
		// pick up pointer to invert index
		dicPtr.withCluster(pkg.cluster());
		LookupPointer leadPtr = memory.lookupDictionaryInvIndex(dicPtr, pkg.valueType());
		LookupPointer allocPtr;

		// Write invert index
		if(leadPtr == null) allocPtr = memory.allocateInvertIndex(Constants.PERUSER, pkg.cluster());
		else allocPtr = leadPtr;

		allocPtr.withCluster(pkg.cluster());
		// System.out.printf("[DEBUG]Pointer: %s ID: %d%n", allocPtr.toString(), pkg.ID());
		LookupPointer invIndexPtr = memory.storeInvertIndex(allocPtr, pkg);

		// refresh dictionary point
		if(leadPtr == null) memory.storeDictionaryToInvIndex(dicPtr, invIndexPtr, pkg.valueType());
	}

	// Look up pointers specified by ID
	private ContentInformation lookupContentPointer(long ID, int subtype, int cluster){
		int i;
		// Lookup pointer by ID
//		System.out.println("[DEBUG] Lookup ID: " + ID);
		ContentInformation currentInfo = memory.lookupContentsSkipPointer(ID, subtype, cluster);
//		System.out.println("[DEBUG] currentInfo: " + (currentInfo == null? "null": currentInfo.getCurrent()));
//		System.out.println("[DEBUG] ID: " + (currentInfo == null? "null": currentInfo.getID()));
		if(currentInfo == null) return null;

		// Incremental search
		LookupPointer previousPointer = LookupPointer.valueOf(-1, -1);
		LookupPointer currentPointer = new LookupPointer(currentInfo.getCurrent()).withCluster(cluster);
		LookupPointer nextPointer = new LookupPointer(currentPointer);
		for(i = 0; i < 100; i++){
			long currentID = memory.lookupContentID(currentPointer);
			nextPointer = memory.lookupContentNext(nextPointer);
			if(nextPointer != null) nextPointer.withCluster(cluster);
			// System.out.printf("[DEBUG:LU1]Prev: %s, Current: %s, Next: %s%n", previousPointer,
			// currentPointer, nextPointer);
			// Not found; ID cannot found
			if(currentID > ID){
				// System.out.printf("[DEBUG:LU1.5]Prev: %s, Current: %s, Next: %s%n", previousPointer,
				// currentPointer, nextPointer);
				return currentInfo.current(null).middle(currentPointer).next(nextPointer).ID(ID)
					.appendSkip(i > Constants.SKIPPTR_INTERVAL);
			}
			if(currentID == ID){
				// System.out.printf("[DEBUG:LU1.7]Prev: %s, Current: %s, Next: %s%n", previousPointer,
				// currentPointer, nextPointer);
				return currentInfo.current(currentPointer).middle(previousPointer).next(nextPointer).ID(ID)
					.appendSkip(i > Constants.SKIPPTR_INTERVAL);
			}
			// Not found; querying ID is newer
			if(nextPointer == null) break;
			previousPointer.copy(currentPointer);
			currentPointer.copy(nextPointer);
			// System.out.printf("[DEBUG:LU2]Prev: %s, Current: %s, Next: %s%n", previousPointer,
			// currentPointer, nextPointer);
		}

		return currentInfo.current(null).behindLast(currentPointer)
			.appendSkip(i > Constants.SKIPPTR_INTERVAL);
	}

	// Store appendix on memory
	public LookupPointer storeAppendix(String content, long ID, int index, int cluster){
		// If any content with same ID is found, it does not over write
		return memory.storeAppendix(AppendixPackage.setPackage().string(content).cluster(cluster), index);
	}

	public void deleteContent(DeleterPackage pkg){
		delFilter.add(pkg);
	}

	public int getDeleteCount(){
		return delFilter.filterCount();
	}

	// ///////////////////////

	public void addCluster(int myOrder, int totalNodes){
		memory.addCluster(myOrder, totalNodes);
	}

	public void optimizeMemory(){
		memory.optimizeMemory();
	}

	public void suspendMyNode(){
		mains.suspendMyNode(0);
	}

	public void resumeMyNode(){
		mains.resumeMyNode();
	}

	public int getClusterBlockAmount(int clusterID){
		return memory.getClusterBlockAmount(clusterID);
	}

	public void setClusterPointInfo(ArrayType info){
		memory.setClusterPointInfo(info);
	}

	public ArrayType getClusterPointInfo(int clusterID){
		return memory.getClusterPointInfo(clusterID);
	}

	public ArrayType toArrayType(int cluster, int block){
		return memory.toArrayType(cluster, block);
	}

	public void fromArrayType(int cluster, int ID, int type, int subtype, byte[] array){
		memory.fromArrayType(cluster, ID, type, subtype, array);
	}

	public void terminate() throws IOException{
		delFilter.terminate();
	}
}
